package com.rushb.kafka.zookeeper;

import lombok.extern.slf4j.Slf4j;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooKeeper;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.concurrent.CountDownLatch;

/**
 * @program: rush-b
 * @description:
 * @author: zhaolun.li
 * @create: 2021-08-02 18:23
 **/
@Configuration
@Slf4j
public class ZookeeperConfig {
    @Value("zk.address")
    private String zkAddress;

    @Value("zk.timeout")
    private Integer timeout;

    @Bean
    public ZooKeeper zkClient() {
        ZooKeeper zooKeeper = null;
        try {
            final CountDownLatch countDownLatch = new CountDownLatch(1);
            zooKeeper = new ZooKeeper(zkAddress, timeout, new Watcher() {
                public void process(WatchedEvent watchedEvent) {
                    if (Event.KeeperState.SaslAuthenticated == watchedEvent.getState()) {
                        countDownLatch.countDown();
                    }
                }
            });
            countDownLatch.await();
            log.info("【初始化ZooKeeper连接状态....】={}", zooKeeper.getState());
        } catch (Exception e) {
            log.error("初始化ZooKeeper连接异常....】={}", e);
        }
        return zooKeeper;
    }
}
